/*
 * Copyright (C) 2017 Intel Corporation.  All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 */

// Copyright (c) WanSheng Intelligent Corp. All rights reserved.


#define LOG_TAG "trans"
#include "agent_core_lib.h"
#include "wasdk_internal_consts.h"

/*typedef*/ struct sync_node_data
{
    void * response;
    int  response_len;
    ptr_sync_t sync_obj;
    uint8_t  status;
    transaction_resp_fmt_t  response_fmt;
};


/*typedef*/ struct async_node_data
{
    void * context_data;
    bh_async_callback cb;
};

typedef struct sync_node
{
    struct sync_node * next;
    //struct sync_node * prev;
    uint32_t id;
    unsigned char sync_type; // 1.sync  2. async
    tick_time_t  timeout;
    union
    {
        struct async_node_data cb_data;
        struct sync_node_data  sync_data;
    };
}sync_node_t;


struct sync_ctx
{
    ptr_sync_t ctx_lock;
    unsigned int cnt;
    void * list;
    uint32_t new_id;
    tick_time_t nearest_expiry_us;
};

//#define PRINTX(...) {printf("%u: ", (uint32_t)bh_get_tick_ms()%100000); printf(__VA_ARGS__);}
#define PRINTX(...)

#ifdef CTX_BUFFER_CECK

// it is used for checking if the user freed the data in the callback function
int g_trans_ctx_malloc_cnt = 0;
void * trans_malloc_ctx(int len)
{
    if(len == 0)
        return NULL;

    g_trans_ctx_malloc_cnt++;
    return malloc_z(len);
}


void   trans_free_ctx(void * ctx)
{
    if(ctx == NULL) return;

    g_trans_ctx_malloc_cnt --;
    free(ctx);
}
#endif


/*
    @brief:  refresh the nearest expiry of whole list
    @return: whether the nearest expiry is changed
*/
static bool bh_refresh_next_expiry(sync_ctx_t sync_ctx)
{
    bh_bsp_lock(sync_ctx->ctx_lock);
    tick_time_t nearest = -1;
    sync_node_t * current = (sync_node_t * )sync_ctx->list;
    while(current)
    {
        if(current->sync_type == T_Async_Trans)
        {
            if(nearest == -1)
            {
                nearest = current->timeout;
            }
            else if(nearest > current->timeout)
            {
                nearest = current->timeout;
            }
        }
        current = current->next;
    }

    bh_bsp_unlock(sync_ctx->ctx_lock);

    if(nearest == -1 && sync_ctx->nearest_expiry_us == -1) 
        return false;

    if(nearest == -1 || sync_ctx->nearest_expiry_us == -1)
    {
        sync_ctx->nearest_expiry_us = nearest;
        return true;
    }

    if(sync_ctx->nearest_expiry_us > nearest)
    {
        sync_ctx->nearest_expiry_us = nearest;
        return true;
    }

    return false;
}


sync_ctx_t  create_sync_ctx()
{
    printf("LOG_MY_MASK_ID=%d\n", LOG_MY_MASK_ID);
    sync_ctx_t ctx = (sync_ctx_t )malloc(sizeof(struct sync_ctx));
    memset((void*) ctx, 0, sizeof (*ctx));
    ctx->new_id = 1;
    ctx->ctx_lock = bh_bsp_create_syncobj();
    TraceD(LOG_FLAG_TRANSACTION, "New transaction context: %p", ctx);
    return ctx;
}


void delete_sync_ctx(sync_ctx_t  ctx)
{
    bool deleted = false;
    bh_bsp_lock(ctx->ctx_lock);
    sync_node_t * current = (sync_node_t * )ctx->list;
    while(current)
    {
        // ensure the callers have chance for cleanup
        if(current->cb_data.cb)
            current->cb_data.cb(current->cb_data.context_data, NULL, 0, 0);     

        if(current->sync_type == T_Sync_Trans)
            bh_bsp_delete_syncobj(current->sync_data.sync_obj);   

        sync_node_t * tmp = current;
        free(tmp);
        deleted = true;
        current = current->next;
    }

    bh_bsp_unlock(ctx->ctx_lock);
    bh_bsp_delete_syncobj(ctx->ctx_lock);

    TraceD(LOG_FLAG_TRANSACTION, "Delete transaction context: %p, deleted=%d", ctx, deleted);

    free(ctx);
}


static void add_sync_node(sync_ctx_t sync_ctx, sync_node_t* node)
{
    bh_bsp_lock(sync_ctx->ctx_lock);

    if(sync_ctx->list == NULL)
    {
        sync_ctx->list = node;
        node->next = NULL;
    }
    else
    {
        node->next = (sync_node_t * ) sync_ctx->list;
        sync_ctx->list = node;
    }

    sync_ctx->cnt ++;

    // update the nearest expiry time
    if(node->sync_type == T_Async_Trans && node->timeout < sync_ctx->nearest_expiry_us)
        sync_ctx->nearest_expiry_us =  node->timeout;

    bh_bsp_unlock(sync_ctx->ctx_lock);

    TraceV(LOG_FLAG_TRANSACTION, "trans[%p]: add node [%p], id=%d, timeout=%dms, total num=%d\n", sync_ctx, node, node->id, node->timeout, sync_ctx->cnt);

}


static sync_node_t * remove_sync_node(sync_ctx_t  sync_ctx, uint32_t id, transaction_type_t type )
{
    bh_bsp_lock(sync_ctx->ctx_lock);
    sync_node_t * prev = NULL;
    sync_node_t * current = (sync_node_t * )sync_ctx->list;
    while(current)
    {
        if(id == current->id && (type == T_Undefined_Sync || type == current->sync_type))
        {
            if(prev)
                prev->next = current->next;
            else
                sync_ctx->list = current->next;

            sync_ctx->cnt --;

            TraceV(LOG_FLAG_TRANSACTION, "trans[%p]: removed node [%p], id=%d, total num=%d\n",sync_ctx, current, current->id, sync_ctx->cnt);
            break;
        }
        else
        {
            prev = current;
            current = current->next;
        }
    }

    bh_bsp_unlock(sync_ctx->ctx_lock);

    return current;

}


static bool find_sync_node(sync_ctx_t  sync_ctx, transaction_id_t id)
{
    bh_bsp_lock(sync_ctx->ctx_lock);
    sync_node_t * current = (sync_node_t * )sync_ctx->list;
    while(current)
    {
        if(id == current->id )
        {
            bh_bsp_unlock(sync_ctx->ctx_lock);
            return true;
        }
        current = current->next;
    }

    bh_bsp_unlock(sync_ctx->ctx_lock);
    return false;
}


/*
 @brief: ensured the generated ID never be 0 and -1
*/
transaction_id_t bh_gen_id(sync_ctx_t ctx)
{
    transaction_id_t id;
    bool exist = true;
    while(exist)
    {
        bh_bsp_lock(ctx->ctx_lock);
        if(ctx->new_id == TRANSACTION_ID_INVALID) ctx->new_id = 1;
        id= ctx->new_id++;
        bh_bsp_unlock(ctx->ctx_lock);

        exist = find_sync_node(ctx, id);
    }
    return id;
}

//
// Note: the node must be deleted in this function for simply design
//
void * bh_wait_response(sync_ctx_t sync_ctx, transaction_id_t id, int * len, int * fmt, uint32_t timeout)
{
    * len = -1;
    void * response = NULL;

    if(id == TRANSACTION_ID_AUTO)
        return NULL;

    if(find_sync_node(sync_ctx, id))
    {
        TraceD(LOG_FLAG_TRANSACTION, "id %u is already present", id);
        return NULL;
    }

    sync_node_t * node = (sync_node_t *)malloc(sizeof(sync_node_t));
    memset((void*) node, 0, sizeof (*node));
    node->id = id;
    node->sync_type = T_Sync_Trans;
    node->sync_data.sync_obj = bh_bsp_create_syncobj();

    // timeout is not really check for the T_Sync_Trans node
    node->timeout = bh_get_tick_us() + timeout *1000;

    add_sync_node(sync_ctx, node);

    // Todo: need ref count to avoid other thread free this node bet
    bh_bsp_lock(node->sync_data.sync_obj);
    bh_bsp_wait(node->sync_data.sync_obj, timeout, 1);

    sync_node_t* node_found = remove_sync_node(sync_ctx, node->id, T_Sync_Trans);
    while( node_found == NULL)
    {
        usleep(1);
        node_found = remove_sync_node(sync_ctx, node->id, T_Sync_Trans);
    }

    assert(node_found == node);

    response = node_found->sync_data.response;

    if(*fmt)
        *fmt = node_found->sync_data.response_fmt;

    *len = node_found->sync_data.response_len;

    bh_bsp_delete_syncobj(node_found->sync_data.sync_obj);
    free(node_found);

    return response;
}

/*
    @brief: handover the response to the waiting party.

    @param  response: pointer to the address of response. 
                      after this function return, its value is reset to NULL.

    @return: true - the buffer was handed over to the waiting party
*/
bool bh_transaction_handover_response(sync_ctx_t  sync_ctx, transaction_id_t id, void ** response, uint8_t format)
{
    sync_node_t * node = remove_sync_node(sync_ctx,id, T_Sync_Trans);
    if(node == NULL)
    {
        return false;
    }
    assert(node->sync_type == T_Sync_Trans);

    bh_bsp_lock(node->sync_data.sync_obj);
    node->sync_data.response_len = 0;
    node->sync_data.response = *response;
    *response = NULL;
    node->sync_data.response_fmt = format;
    add_sync_node(sync_ctx, node);

    // call sequence important here: ensure
    bh_bsp_wakeup(node->sync_data.sync_obj, 0);

    return true;
}

/*
 @return: true - success
 @param len:  0 - the ownership of response is handled over to the callback. 
                   Normally the waiter do the release work.
              >0 - the content of response will be copied. 
*/
bool bh_feed_response(sync_ctx_t sync_ctx, transaction_id_t id, void * response,  uint32_t len, transaction_resp_fmt_t format)
{
    //printf("In bh_feed_response. mid=%d, response=%s, len=%d, fmt=%d\n", id, response, len, format);
    sync_node_t * node = remove_sync_node(sync_ctx,id, T_Undefined_Sync);
    if(node == NULL)
    {
        TraceV(LOG_FLAG_DEBUG_TRANS,"bh_feed_response, node == NULL, transID = %d\n", id);
        return false;
    }
    if(node->sync_type == T_Sync_Trans)
    {
        bh_bsp_lock(node->sync_data.sync_obj);
        node->sync_data.response_len = 0;
        node->sync_data.response = response;
        node->sync_data.response_fmt = format;

        if(len != 0)
        {
            node->sync_data.response = malloc(len);
            if(node->sync_data.response == NULL)
            {
                TraceD(LOG_FLAG_TRANSACTION, "malloc failed. len=%d, transID=%d", len, id);
            }
            else
            {
                node->sync_data.response_len = len;
                memcpy(node->sync_data.response, response, len);
            }
        }

        add_sync_node(sync_ctx, node);

        // call sequence important here: ensure
        bh_bsp_wakeup(node->sync_data.sync_obj, 0);
        return true;
    }
    else if(node->sync_type == T_Async_Trans)
    {
        // the work thread api should post the callback to the target thread
        node->cb_data.cb(node->cb_data.context_data, response, len, format);

        // check and update the nearest expiry
        if(node->timeout == sync_ctx->nearest_expiry_us)
        {
            bh_refresh_next_expiry(sync_ctx);
        }
        free(node);
        return true;
    }

    return false;
}


uint32_t bh_get_next_expiry_ms(sync_ctx_t sync_ctx)
{
    if(sync_ctx->nearest_expiry_us == -1)
        return -1;

    tick_time_t now_us = bh_get_tick_us();
    if(now_us <=  sync_ctx->nearest_expiry_us)
        return (sync_ctx->nearest_expiry_us - now_us) / 1000;

    return 0;
}

/***
  @param ctx_data:  The user data that will be passed into the callback. 
                    Its buffer should be allocated by the caller and released by callback. 
  @return:          The id of the new transaction node
***/
transaction_id_t bh_wait_response_async(sync_ctx_t sync_ctx, transaction_id_t id,
        bh_async_callback cb,
        void* ctx_data,
        uint32_t timeout,
        void * worker_thread)
{
    UNUSE(worker_thread);

    if(cb == NULL) return TRANSACTION_ID_INVALID;

    if(id == TRANSACTION_ID_AUTO)
    {
        id = bh_gen_id(sync_ctx);
    }
    else if(find_sync_node(sync_ctx, id))
    {
        TraceD(LOG_FLAG_TRANSACTION, "id %u is already present", id);
        return TRANSACTION_ID_INVALID;
    }
   

    tick_time_t now_us = bh_get_tick_us();
    sync_node_t * node = (sync_node_t *)malloc(sizeof(sync_node_t));
    if(node == NULL) return TRANSACTION_ID_INVALID;
    memset((void*) node, 0, sizeof (*node));
    node->id = id;
    node->sync_type = T_Async_Trans;
    node->cb_data.context_data = ctx_data;
    node->cb_data.cb = cb;
    node->timeout = timeout*1000 + now_us;
    add_sync_node(sync_ctx, node);

    return node->id;
}


transaction_id_t bh_transaction_new(sync_ctx_t sync_ctx, 
        bh_async_callback cb, void* ctx_data, uint32_t timeout_ms)
{
    return bh_wait_response_async(sync_ctx, TRANSACTION_ID_AUTO, cb, ctx_data, timeout_ms, NULL);
}

/*
  @brief: remove all expired nodes to the expired_list
  @return: the *microsecond* to the nearest expiry
*/
static uint64_t bh_remove_expired_trans(sync_ctx_t sync_ctx, sync_node_t ** expired_list)
{
    #define MOVE_NEXT(prev, current) {prev = current; current = current->next;}

    bh_bsp_lock(sync_ctx->ctx_lock);
    tick_time_t now_us = bh_get_tick_us();
    if(now_us < sync_ctx->nearest_expiry_us)
    {
        bh_bsp_unlock(sync_ctx->ctx_lock);
        return (sync_ctx->nearest_expiry_us - now_us);
    }
  
    tick_time_t nearest = -1;

    sync_node_t * expired = NULL;
    sync_node_t * prev = NULL;
    sync_node_t * current = (sync_node_t * )sync_ctx->list;
    while(current)
    {
        if(current->sync_type == T_Sync_Trans)
        {
            MOVE_NEXT(prev, current);
        }
        else if(current->timeout > now_us)  // not expiried
        {
            if(nearest == -1)
            {
                nearest = current->timeout;
            }
            else if(nearest > current->timeout)
            {
                nearest = current->timeout;
            }

            MOVE_NEXT(prev, current);
        }
        else  // expired
        {
            sync_node_t * sv_node = current->next;
            if(prev)
                prev->next = current->next;
            else
                sync_ctx->list = current->next;

            current->next = expired;
            expired = current;

            current=sv_node;
            sync_ctx->cnt --;

            //TraceV(LOG_FLAG_TRANSACTION,  "trans[%p]: expired, removed node [%p], id=%d, total num=%d\n",sync_ctx, expired, expired->id, sync_ctx->cnt);
        }
    }

    sync_ctx->nearest_expiry_us = nearest;
    bh_bsp_unlock(sync_ctx->ctx_lock);

    if(expired_list) *expired_list = expired;

    if(nearest == -1) return -1;
    return (nearest - now_us);
}

/*
  @brief: execute callbacks for all expired nodes
  @return: the millisecond to the nearest expiry
*/
uint32_t bh_handle_expired_trans(sync_ctx_t sync_ctx)
{
    sync_node_t * expired_list = NULL;
    uint64_t timeout = bh_remove_expired_trans(sync_ctx, &expired_list);

    while(expired_list)
    {
        sync_node_t *node = expired_list;
        expired_list = expired_list->next;
        TraceI(LOG_FLAG_DEBUG_TRANS, "trans [%p]: handle expired node [%p], transID=%d, next=[%p]\n",sync_ctx, node, node->id, node->next);

        if(node->sync_type == T_Async_Trans)
        {
            node->cb_data.cb(node->cb_data.context_data, NULL, 0, 0);
        }

        if(node)
            free(node);
    }

    if(timeout == -1) return -1;

    uint32_t timeout_ms = timeout / 1000;
    // it might be 0 ms after the conversion from microsecond to millisecond
    if(timeout_ms == 0) 
        timeout_ms = 1;

    return timeout;
}


